package cn.cmft.jc.transform;

import cn.cmft.jc.transform.joiner.ImpactTask;
import cn.cmft.jc.transform.joiner.ImpactTaskResult;
import cn.cmft.jc.transform.parse.ChangeDetail;
import cn.cmft.jc.transform.parse.ChangeOperationType;
import com.alibaba.fastjson2.JSONObject;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.util.Collector;
import java.util.List;

public class RecordFormat extends RichFlatMapFunction<ImpactTaskResult, String> {
    @Override
    public void flatMap(ImpactTaskResult impactTaskResult, Collector<String> collector) {
        String host = impactTaskResult.getHost();
        int port = impactTaskResult.getPort();
        String dbName = impactTaskResult.getDbName();
        String tableName = impactTaskResult.getTableName();
        Long tsMs = impactTaskResult.getTsMs();
        String ddl = impactTaskResult.getDdl();
        List<ImpactTask> impactTasks = impactTaskResult.getImpactTasks();
        for (ImpactTask task: impactTasks){
            ChangeOperationType operationType = task.getOperationType();
            String taskName = task.getTask_name();
            List<ChangeDetail> changeDetail = task.getChangeDetail();
            for (ChangeDetail detail:changeDetail){
                JSONObject root = new JSONObject();
                root.put("host",host);
                root.put("port",port);
                root.put("db_name",dbName);
                root.put("table_name",tableName);
                root.put("ddl",ddl);
                root.put("ts_ms",tsMs);
                root.put("operation_type",operationType);
                root.put("task_name",taskName);
                root.put("old_table_name",detail.getOldTableName());
                root.put("current_table_name",detail.getCurrentTableName());
                root.put("old_column_name",detail.getOldColumnName());
                root.put("current_column_name",detail.getCurrentColumnName());
                root.put("current_column_type",detail.getCurrentColumnType());
                collector.collect(root.toJSONString());
            }
        }
    }
}
